package conductor

import (
    "cing.io/cing-api/task"
    "context"
    "google.golang.org/protobuf/types/known/timestamppb"
    "math/rand"
    "time"

    log "github.com/sirupsen/logrus"
    "k8s.io/apimachinery/pkg/util/sets"

    "cing.io/cing-api/planner"
    "cing.io/cing-api/singer"
    "cing.io/cing/internal/clientcache"
    "cing.io/cing/internal/config"
    "cing.io/cing/internal/storage"
    "cing.io/cing/pkg/util"
)

func Conduct(plan *planner.SingingPlan) {
    tsk := storage.GetTask(util.SplitNamespaceAndName(plan.TaskFullName))
    pc := clientcache.Planner.GetPlanner(tsk.TaskType)

    log.Debug("conduct plan...")

    set := clientcache.Singer.GetSingerSet(util.SplitNamespaceAndName(plan.TaskFullName))
    size := len(set)

    switch plan.SingerSelector.(type) {
    case *planner.SingingPlan_SelectAllSinger:
        for ip, cli := range set {
            go innerConduct(cli, pc, tsk, plan, ip)
        }

    case *planner.SingingPlan_SelectNoSinger:
    case *planner.SingingPlan_SelectOneRandomSinger:
        r := rand.Intn(size)
        for ip, cli := range set {
            r--
            if r <= 0 {
                innerConduct(cli, pc, tsk, plan, ip)
            }
        }

    case *planner.SingingPlan_SelectSharding:
        selector := plan.SingerSelector.(*planner.SingingPlan_SelectSharding)
        shardCnt := selector.SelectSharding.ShardCnt
        shardNumList := selector.SelectSharding.ShardNumList

        var shardSize int
        if size%int(shardCnt) == 0 {
            shardSize = size / int(shardCnt)
        } else {
            shardSize = size/int(shardCnt) + 1
        }

        shardCursor := 0
        shardNum := shardNumList[shardCursor]
        shardBase := int(shardNum-1)*shardSize + 1
        i := 1
        for ip, cli := range set {
            if i >= shardBase {
                if i < shardBase+shardSize {
                    go innerConduct(cli, pc, tsk, plan, ip)
                } else {
                    shardCursor++
                    shardNum = shardNumList[shardCursor]
                    shardBase = int(shardNum-1)*shardSize + 1
                }
            }
            i++
        }

    case *planner.SingingPlan_SelectSpecificSinger:
        selector := plan.SingerSelector.(*planner.SingingPlan_SelectSpecificSinger)
        ipList := selector.SelectSpecificSinger.SingerIpList
        ipSet := sets.NewString(ipList...)
        for ip, cli := range set {
            if ipSet.Has(ip) {
                go innerConduct(cli, pc, tsk, plan, ip)
            }
        }

    case *planner.SingingPlan_SelectTheFirstSinger:
        for ip, cli := range set {
            innerConduct(cli, pc, tsk, plan, ip)
            break
        }
    }
}

func innerConduct(cli singer.CingSingerClient, pc planner.CingPlannerClient,
    tsk *task.Task, plan *planner.SingingPlan, ip string) {

    log.Debug("inner conduct plan, ip is: ", ip)

    timeout := config.ConductTimeout
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()

    log.Debugf("Grpc call Time: %s", time.Now())

    // Contact the server and print out its response.
    resp, err := cli.Singing(ctx, &singer.SingingCommandRequest{
        TaskFullName: plan.TaskFullName,
        StartTime:    plan.StartTime,
        SingerIp:     ip,
    })
    if err != nil {
        log.Errorf("Error: %v", err)
    }

    log.Debugf("Grpc resp Time: %s", time.Now())
    log.Debugf("Grpc Response: %s", resp.String())

    if resp == nil {
        log.Error("Grpc call error, response is nil!")
        return
    }

    switch resp.ExecuteStatus {
    case singer.SingingCommandResponse_OK:
        // send OnSingingStart event
        resp, err := pc.OnSingingStart(context.TODO(), &planner.OnSingingStartRequest{
            Task:     tsk,
            ThisPlan: plan,
            SingerIp: ip,
        })
        if err != nil {
            log.Error("call OnSingingStart hook error: ", err)
        }

        // put new error
        if resp.NextPlan != nil {
            err = PutPlan(resp.NextPlan)
            if err != nil {
                log.Error("put new plan error: ", err)
            }
        }

        // update current plan status to "DOING"
        err = storage.UpdatePlanStatus(plan.TaskFullName, plan.StartTime.AsTime().UnixMilli(), planner.SingingPlan_DOING)
        if err != nil {
            log.Error("conductor update plan status error: ", err)
        }
    case singer.SingingCommandResponse_REJECT:
        log.Errorf("conductor error: %s, plan info: %s", singer.SingingCommandResponse_REJECT.String(), plan.String())
        resp, err := pc.OnSingingEnd(context.TODO(), &planner.OnSingingEndRequest{
            Task:            tsk,
            ThisPlan:        plan,
            SingerIp:        ip,
            Successful:      false,
            ActualStartTime: timestamppb.Now(),
            ActualEndTime:   timestamppb.Now(),
            Feedback:        nil,
        })
        if err != nil {
            log.Error("call OnSingingEnd hook error: ", err)
        }
        err = storage.DeletePlan(util.GetTaskFullName(tsk), plan.StartTime.AsTime().UnixMilli())
        if err != nil {
            log.Error("delete task error: ", err)
        }
        if resp.NextPlan != nil {
            err = PutPlan(resp.NextPlan)
            if err != nil {
                log.Error("put new plan error: ", err)
            }
        }
    }

}
